package chapter12.eg2;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

public class HeartBeatReqHandler extends SimpleChannelInboundHandler<NettyMessage> {
  private volatile ScheduledFuture<?> heartBeat;

  @Override
  protected void channelRead0(ChannelHandlerContext ctx, NettyMessage message) throws Exception {
    if (message.getHeader() != null
      && message.getHeader().getType() == MessageType.LOGIN_RESP.value()) {
      heartBeat = ctx.executor()
        .scheduleAtFixedRate(new HeartBeatTask(ctx), 0, 5000, TimeUnit.MILLISECONDS);
    } else if (message.getHeader() != null
      && message.getHeader().getType() == MessageType.HEARTBEAT_RESP.value()) {
      System.out.println("Client receive server heart beat message: --> " + message);
    } else {
      ctx.fireChannelRead(message);
    }
  }

  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    if (heartBeat != null) {
      heartBeat.cancel(true);
      heartBeat = null;
    }

    ctx.fireExceptionCaught(cause);
  }

  private NettyMessage buildHeartBeat() {
    NettyMessage message = new NettyMessage();
    Header header = new Header();
    header.setType(MessageType.HEARTBEAT_REQ.value());
    message.setHeader(header);
    return message;
  }

  private class HeartBeatTask implements Runnable {
    private final ChannelHandlerContext ctx;

    public HeartBeatTask(final ChannelHandlerContext ctx) {
      this.ctx = ctx;
    }

    @Override
    public void run() {
      NettyMessage heartBeat = buildHeartBeat();
      System.out.println("Client send heart beat message to server: --> " + heartBeat);
      ctx.writeAndFlush(heartBeat);
    }
  }
}
